package com.gitee.ywj1352.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * 消费者监听
 * 生产者
 */
@Component
public class YWJKfakaConsumer {

    @KafkaListener(topics = {"ywj_test_kafka"},groupId = "consumer1")
    public void listen(List<ConsumerRecord> records, Acknowledgment ack){

        System.out.println("::::::消费1::::::");
        records.forEach(r->{
            System.out.println("消费1"+":"+r.topic()+":"+r.value());
        });
       // ack.acknowledge();
    }

    @KafkaListener(topics = {"ywj_test_kafka"}, groupId = "consumer2")
    public void listen2(List<ConsumerRecord> records, Acknowledgment ack){

        System.out.println("::::::消费2::::::");
        records.forEach(r->{
            System.out.println("消费2"+":"+r.topic()+":"+r.value());
        });
         ack.acknowledge();
    }
}
